package com.jun.admin.core.listener;

import com.jun.admin.core.model.constant.RedisKeyConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author yuejun.li
 * @date 2024/8/13 10:51:02
 */
@Slf4j
//@Component
public class RedisListener implements ApplicationRunner,Runnable {

    @Autowired
    private RedisTemplate redisTemplate;

    private ExecutorService executorService;

    @PostConstruct
    public void init(){
        executorService = Executors.newSingleThreadExecutor();
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        /*Thread thread = new Thread(this);
        thread.setName("consumer-thread");
        thread.start();*/
        executorService.execute(this);
    }

    @Override
    public void run() {
        //listMQ();
        zsetMQ();

    }

    public void zsetMQ() {
        // 创建SortedSet实例
        ZSetOperations zSetOperations = redisTemplate.opsForZSet();
        while (true){
            // 拿取数据 (rangeByScore返回有序集合中指定分数区间的成员列表。有序集成员按分数值递增(从小到大)次序排列)
            Set<String> order = zSetOperations.rangeByScore(RedisKeyConstant.REDIS_ZSET_QUEUE_KEY, 0, System.currentTimeMillis(), 0, 1);
            if (ObjectUtils.isEmpty(order)) {
                try {
                    //log.info("当前没有数据 当前线程睡眠5秒");
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 跳过本次循环 重新循环拿取数据
                continue;
            }

            // 利用迭代器拿取Set中的数据
            String massage = order.iterator().next();
            if (zSetOperations.remove(RedisKeyConstant.REDIS_ZSET_QUEUE_KEY, massage) > 0) {
                //做些业务处理
                log.info("我拿到的消息：{}" + massage);
            }
        }
    }

    public void listMQ() {
        while(true){
            Object item = redisTemplate.opsForList().rightPopAndLeftPush(RedisKeyConstant.REDIS_QUEUE_KEY, RedisKeyConstant.REDIS_PENDING_QUEUE_KEY, 5, TimeUnit.SECONDS);

            if(item == null){
                continue;
            }

            try{
                log.info("消费成功：{}",item);
                redisTemplate.opsForList().remove(RedisKeyConstant.REDIS_PENDING_QUEUE_KEY,0,item);
            }catch (Exception e){
                break;
            }
        }
    }
}
